package updown.service.updown;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.stereotype.Service;
import updown.constant.GeneralConstant;
import updown.constant.MsgConstant;
import updown.constant.MyInstanceConstant;
import updown.data.ExecutorPoolUtil;
import updown.data.SseCacheData;
import updown.domain.*;
import updown.service.jar.SshService;
import updown.service.jar.UploadJarService;
import updown.service.task.ServerHostService;
import updown.util.DateUtils;
import updown.util.MsgUtil;
import updown.util.SleepUtil;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

/**
 * <p>Title: ww.kaixin.downupservice.serviceInfo </p>
 * <p>Description:updown: - 重启服务 serviceInfo  </p>
 *
 * @Author:jiangkaixin
 * @Date:2021/11/3
 * @Version: 1.0
 */
@Slf4j
@Service
public class DownUpServiceWithRetry {
    @Autowired
    private SseCacheData sseCacheData;

    @Autowired
    private SshService sshService;

    @Autowired
    private ServerHostService serverHostService;

    @Autowired
    private EurekaService eurekaService;

    @Autowired
    private LocalService localService;

    public static final Map<String, Set<String>> downDownMap = new HashMap<>();

    /**
     * 重写 12-20
     *
     * @param taskWrapper
     */
    public void executeInPool(TaskWrapper taskWrapper) {
        ExecutorPoolUtil.executor.execute(() -> {
            SleepUtil.sleepSeconds(1); //为了 sse
            try {
                if (executeDownUp(taskWrapper)) {
                    sseCacheData.sendLastMsgAndComplete(MsgConstant.down_up_finish + taskWrapper.spendTime() + "秒", taskWrapper);
                }
            } catch (Exception e) {
                log.info("executeDownUp() 方法报错:{}", e);
                sseCacheData.sendLastMsgAndThrowEx(e.toString(), taskWrapper);
            }
        });
    }

    /**
     * -- 第一阶段，各个client 上的 缓存被清除了：（发送下线之后休息 90s）
     * 1. 00:00 发送下线请求,清除了 readWrite 中的所有，但是 read 中还有(脏数据)
     * 2. 00:29 （30s/的拉取操作）直接从 readMap 获取到过期的数据，还是UP状态 （1. 配置同步间隔时间）
     * 3. 00:30 readMap 和 readWrite 完成同步操作
     * 4. 00:58 的时候刚好完成本地 instances 集合 到Ribbon 中的更新 (脏数据)
     * 5. 00:59 再次拉取到更新完成之后的 readMap 数据：本地数据更新
     * 6. 01:28 ribbon中过滤掉 DOWN 掉的节点 不会再调用了。（1. 配置本地的同步时间； 2. 自定义实现类的方式，事件通知更新）
     * -- 第二
     * 7. 01:30 发送 kill -15 操作： 执行 DiscoveryClient#@PreDestroy#shutdown() 操作 大概耗时 10s；【假设10s】
     * -- 第三阶段
     * 8. 01:40 重新启动： 注册到 Server 假设30s 【假设30s】
     */
    public Boolean executeDownUp(TaskWrapper taskWrapper) {
        log.info("执行上下线的线程name:{}", Thread.currentThread().getName());
        String serviceName = taskWrapper.getServiceInfo().getServiceName();
        // 1 是否替换检查、jar和脚本 检查、是否正在替换检查、本地和eureka中服务信息检查
        doCheck(taskWrapper);
        // 2 上面已经检查过了，这里直接返回 instance 信息
        List<ServiceInstance> instances = localService.getLocalInstancesInfo(taskWrapper);
        Set<String> instanceIdSet = instances.stream().map(instance -> instance.getInstanceId()).collect(Collectors.toSet());
        downDownMap.put(serviceName, instanceIdSet);

        if (instanceIdSet.size() == 1){
            log.info("只发现一个服务实例，直接执行 killscript 脚本");
            sseCacheData.pushMsg("只发现一个服务实例，直接执行 killscript 脚本", taskWrapper);
            String hostOrIP = instances.get(0).getHost();
            ServerHost host = serverHostService.getServerHostNameOrIP(hostOrIP);
            TaskInfo kill9Task = sshService.kill9(host, taskWrapper);
            taskWrapper.getTaskInfos().add(kill9Task);
        }else {
            int number = 1;
            for (ServiceInstance instance : instances) {
                log.info("开始执行服务替换，第【:{}】个，当前的服务实例信息是:{}", number, instance);
                String hostOrIP = instance.getHost();
                ServerHost serverHost = serverHostService.getServerHostNameOrIP(hostOrIP);
                DownUpRetry downUpTask = DownUpRetry.builder()
                        .currentInstanceId(instance.getInstanceId())
                        .serviceName(serviceName)
                        .serverHost(serverHost)
                        .build();
                taskWrapper.setDownUpRetry(downUpTask);
                // 3 下线
                try {
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_SEND_DOWN);
                    // 4 准备 发送 kill - 15 请求 ： 等待 15s
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_KILL15);
                    // 5 检查 是否真的下线了
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_CHECK_CANCEL);
                    // 6 执行重启脚本
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_RESTART);
                    // 7 检查是否注册到eureka 中
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_RE_REGISTER_EUREKA);
                    // 8 检查 本地的discovery 缓存有无 重新注册的服务信息
                    retryFunc.apply(taskWrapper, GeneralConstant.RETRY_TYPE_CHECK_LOCAL);
                } catch (Exception e) {
                    log.info("345678 错误:{}", e);
                }
                // 9 执行下一个替换
                instanceIdSet.remove(instance.getInstanceId());
                if (instanceIdSet.size() > 0) {
                    sseCacheData.pushMsg(MsgConstant.ready_to_start_next, taskWrapper);
                    log.info("log.info: 开始执行下一个服务替换");
                }
            }
        }
        downDownMap.remove(serviceName);
        return Boolean.TRUE;
    }

    /**
     * 重试 方法
     */
    BiFunction<TaskWrapper, String, Boolean> retryFunc = (taskWrapper, type) -> {
        BiFunction<TaskWrapper, Integer, Boolean> func = getRetryFunc(type);
        Integer retryTimes = taskWrapper.getDownUpRetry().getRetryTimes(type);
        Integer retryPeriodSec = taskWrapper.getDownUpRetry().getRetryPeriod(type);
        Integer retryCount = 0;
        while (retryTimes >= 0) {
            if (func.apply(taskWrapper, retryCount)) {
                return true;
            }
            SleepUtil.sleepSeconds(retryPeriodSec);
            retryTimes--;
            retryCount++;
        }
        sseCacheData.sendLastMsgAndThrowEx(MsgConstant.getFailMsgByType(type), taskWrapper);
        return false;
    };

    /**
     * @param type
     * @return
     */
    BiFunction<TaskWrapper, Integer, Boolean> getRetryFunc(String type) {
        if (GeneralConstant.RETRY_TYPE_KILL15.equals(type)) {
            return kill15Func;
        } else if (GeneralConstant.RETRY_TYPE_CHECK_CANCEL.equals(type)) {
            return checkCancelFunc;
        } else if (GeneralConstant.RETRY_TYPE_SEND_DOWN.equals(type)) {
            return sendDown;
        } else if (GeneralConstant.RETRY_TYPE_RESTART.equals(type)) {
            return restartScriptFunc;
        } else if (GeneralConstant.RETRY_TYPE_RE_REGISTER_EUREKA.equals(type)) {
            return reRegisterInEureka;
        } else if (GeneralConstant.RETRY_TYPE_CHECK_LOCAL.equals(type)) {
            return checkLocalFunc;
        }
        return null;
    }

    /**
     * 阶段一  发送down 请求
     */
    BiFunction<TaskWrapper, Integer, Boolean> sendDown = (taskWrapper, tryCount) -> {
        if (tryCount == 0) {
            sseCacheData.pushMsg(MsgConstant.send_down_msg, taskWrapper);
        }
        String instanceId = taskWrapper.getDownUpRetry().getCurrentInstanceId();
        String serviceName = taskWrapper.getDownUpRetry().getServiceName();
        // 1 发送 status=DOWN 下线请求
        TaskInfo downTask = eurekaService.sendDown(instanceId, serviceName);
        SleepUtil.sleepSeconds(1);
        Boolean isDown = eurekaService.isSameStatus(serviceName, instanceId, MyInstanceConstant.DOWN);
        if (!isDown) {
            taskWrapper.getTaskInfos().add(downTask.finishedUnSuccess());
        } else {
            log.info("DOWN状态检测一致");
            taskWrapper.getTaskInfos().add(downTask.finished());
        }
        if (downTask.getSuccess()) {
            sseCacheData.pushMsg(MsgUtil.task2Msg(downTask), taskWrapper);
            log.info("DOWN状态检测一致,准备睡眠等待缓存清除");
            // 睡眠 90s
            sseCacheData.pushMsg(MsgConstant.send_down_msg_after_sleep, taskWrapper);
            SleepUtil.sleepSeconds(90);
        }
        return downTask.getSuccess();
    };

    /**
     * 阶段二  kill 15
     */
    BiFunction<TaskWrapper, Integer, Boolean> kill15Func = (taskWrapper, tryCount) -> {
        if (tryCount == 0) {
            sseCacheData.pushMsg(MsgConstant.ready_to_kill_15, taskWrapper);
        }
        TaskInfo taskInfo = sshService.execKill15(taskWrapper.getDownUpRetry().getServerHost(), taskWrapper);
        taskWrapper.getTaskInfos().add(taskInfo);
        if (taskInfo.getSuccess()) {
            sseCacheData.pushMsg(MsgConstant.kill_15_script_success, taskWrapper);
            log.info("kill15执行成功，等待30s，检测是否下线成功");
            SleepUtil.sleepSeconds(30);
        }
        return taskInfo.getSuccess();
    };

    /**
     * 阶段二 检查 是否cancel 成功的任务
     */
    BiFunction<TaskWrapper, Integer, Boolean> checkCancelFunc = (taskWrapper, tryCount) -> {
        TaskInfo taskInfo = TaskInfo.builder().taskDes("去Eureka检查是否下线成功").build();
        MyInstanceInfo cancelInstanceInfo = eurekaService.getSpecificInstanceByNameAndInstanceId(taskWrapper.getServiceInfo().getServiceName()
                , taskWrapper.getDownUpRetry().getCurrentInstanceId());
        if (cancelInstanceInfo == null) {
            log.info("执行kill15 过15s 之后 eureka 中已无此实例信息");
            taskWrapper.getTaskInfos().add(taskInfo.finished());
        } else {
            log.info("kill 15 之后 eureka 中拉取到的 server 信息:{}", cancelInstanceInfo);
            taskWrapper.getTaskInfos().add(taskInfo.finishedUnSuccess());
        }
        if (taskInfo.getSuccess()) {
            sseCacheData.pushMsg(MsgConstant.kill_15_success, taskWrapper);
        }
        return taskInfo.getSuccess();
    };

    /**
     * 阶段三 重启脚本
     */
    BiFunction<TaskWrapper, Integer, Boolean> restartScriptFunc = (taskWrapper, tryCount) -> {
        // 3 执行重新启动脚本
        if (tryCount == 0) {
            sseCacheData.pushMsg(MsgConstant.start_script_ready, taskWrapper);
        }
        ServerHost serverHost = taskWrapper.getDownUpRetry().getServerHost();
        TaskInfo startTask = TaskInfo.builder().startTime(DateUtils.getCurDateTime()).taskDes("重启服务脚本").build();
        if (sshService.execStart(serverHost, taskWrapper).getSuccess()) {
            sseCacheData.pushMsg(MsgConstant.start_script_success, taskWrapper);
            log.info("执行start成功，准备睡眠30s");
            SleepUtil.sleepSeconds(30);
            taskWrapper.getTaskInfos().add(startTask.finished());
        }
        return startTask.getSuccess();
    };

    /**
     * 阶段三 eureka 中是否有实例信息
     */
    BiFunction<TaskWrapper, Integer, Boolean> reRegisterInEureka = (taskWrapper, tryCount) -> {
        // 检查 重新注册到 server 端的状态
        log.info("睡眠30s之后，检查是否重新注册到 eureka 中");
        String instanceId = taskWrapper.getDownUpRetry().getCurrentInstanceId();
        String serviceName = taskWrapper.getDownUpRetry().getServiceName();
        TaskInfo reRegisterInEurekaTask = TaskInfo.builder().startTime(DateUtils.getCurDateTime()).taskDes("检查是否成功重启，注册到eureka").build();
        if (eurekaService.isRestartSuccess(serviceName, instanceId, 0)) {
            log.info("重新注册到 eureka 中，准备睡眠 90s，等待节点同步");
            sseCacheData.pushMsg(MsgConstant.start_to_notify_all_service, taskWrapper);
            taskWrapper.getTaskInfos().add(reRegisterInEurekaTask.finished());
            SleepUtil.sleepSeconds(90);
        } else {
            log.info(MsgConstant.start_script_success_but_not_register_in_eureka_server);
            reRegisterInEurekaTask.finishedUnSuccessAddDes("当前是第【" + (tryCount + 1) + "】从尝试");
        }
        return reRegisterInEurekaTask.getSuccess();
    };

    /**
     * 阶段四 本地是否有 重新注册的缓存
     */
    BiFunction<TaskWrapper, Integer, Boolean> checkLocalFunc = (taskWrapper, tryCount) -> {
        if (tryCount == 0) {
            log.info("检查 本地的discovery 缓存有无 重新注册的服务信息");
            sseCacheData.pushMsg(MsgConstant.start_to_notify_all_service_check, taskWrapper);
        }
        TaskInfo startRecheckTask = TaskInfo.builder().startTime(DateUtils.getCurDateTime()).taskDes("服务重启,本地服务确认工作").build();
        String instanceId = taskWrapper.getDownUpRetry().getCurrentInstanceId();
        String serviceName = taskWrapper.getDownUpRetry().getServiceName();
        if (eurekaService.checkLocalRegistryHasInstanceInfo(serviceName, instanceId)) {
            log.info("本地的discovery 成功检查到  重新注册的服务信息");
            taskWrapper.getTaskInfos().add(startRecheckTask.finished());
            sseCacheData.pushMsg(MsgConstant.local_service_has_instance, taskWrapper);
        } else {
            taskWrapper.getTaskInfos().add(startRecheckTask.finishedUnSuccess());
            sseCacheData.pushMsg(MsgConstant.local_service_has_instance_fail, taskWrapper);
        }
        return startRecheckTask.getSuccess();
    };

    /**
     * 检查工作：
     * 1. 是否正在替换 是否正在上传
     * 2. jar包信息 和 脚本完整度
     * 3. 服务是否在eureka 中存在
     *
     * @param taskWrapper
     */
    private void doCheck(TaskWrapper taskWrapper) {
        doCheckIfDownUpIngAndJarUploadCheckWork(taskWrapper);
        sshService.doCheckJarAndScript(taskWrapper);
        localService.doCheckLocalAndEurekaSame(taskWrapper);
        log.info("检查工作完成，可以替换");
    }

    /**
     * 是否正在替换 是否正在上传
     *
     * @param taskWrapper
     */
    private void doCheckIfDownUpIngAndJarUploadCheckWork(TaskWrapper taskWrapper) {
        // 1 是否正在替换
        if (downDownMap.get(taskWrapper.getServiceInfo().getServiceName()) != null) {
            log.info("正在替换中，本地替换终止");
            sseCacheData.sendLastMsgAndThrowEx("此服务正在替换中请勿重复替换", taskWrapper);
        }
        log.info("未在替换中，本次可以替换");
        sseCacheData.pushMsg("检查1：服务未在替换中", taskWrapper);

        // 2 是否正在上传jar
        log.info("检查 是否正在上传 jar 包 ()");
        if (UploadJarService.sendToSeverIng.get(taskWrapper.getServiceInfo().getServiceName()) != null) {
            log.info("正在jar 同步中，本地替换终止");
            sseCacheData.sendLastMsgAndThrowEx("当前服务实例，正在同步Jar包信息到对应的服务器节点中，无法完成替换操作", taskWrapper);
        }
        log.info("未在jar同步中中，本次可以替换");
        sseCacheData.pushMsg("检查3：未在 jar 包上传中", taskWrapper);
    }
}
